//package cn.ipanel.bigdata.example
//
//import cn.ipanel.bigdata.job.basic
//import org.apache.spark.sql.{DataFrame, Row, SparkSession}
//import org.apache.spark.sql.functions.{col, lit}
//import org.apache.spark.sql.types.LongType
//
///**
// * @author lzz
// * @environment IntelliJ IDEA 2020.3.1
// * @projectName bigdata_shanxi_git
// * @date 2023/06/16 14:56
// * @description:
// */
//object RedisTest {
//
//  def main(args: Array[String]): Unit = {
//    val spark = SparkSession
//      .builder()
//      .appName("myApp")
//      .master("local[*]")
//      .config("spark.redis.host", "192.168.37.161")
//      .config("spark.redis.port", "6379")
//      .config("spark.redis.auth", "ipanel")
////      .config("spark.redis.db", 0)
//      .config("spark.redis.db", 1)
//      .getOrCreate()
//
////    loadFromRedis(spark)
////    loadFromRedis2(spark)
//    val df = loadFromRedis3(spark)
//    df.show(false)
////    saveToReids(spark)
//  }
//
//  def loadFromRedis3(spark: SparkSession): DataFrame = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    val rdd = sc.fromRedisZSetWithScore("test", 3)
//    rdd.toDF("stb_id", "timestamp")
//      .union(
//        spark.sparkContext.parallelize(Seq(
//          ("003501FF0001181001FC08A5C8CADD4F", "123"),
//          ("003501FF0001181001FC08A5C8D14093", "123")
//        )).toDF("stb_id", "timestamp")
//      )
//  }
//
//
//  def saveToReids(spark: SparkSession): Unit = {
//    import spark.implicits._
//    val rdd = spark.sparkContext.parallelize(Seq(
//      ("key1", "100"),
//      ("key2", "101"),
//      ("key3", "50")
//    ))
//
//    // 这种存储方式，将产生n多不认识的key
//    // 1) "foo:c16d89b53a194ddd831e45a60e0d574d"
//    // 2) "_spark:foo:schema"
//    // 3) "foo:ba8626441fd5474f9164a9585cf5b983"
//    // 4) "foo:64f61b647799494ebcd35e7c63ca074f"
//    //    df.write
//    //      .format("org.apache.spark.sql.redis")
//    //      .option("table", "foo")
//    //      .save()
//
//    import com.redislabs.provider.redis._
//    // 效果还是可以的
//    // 127.0.0.1:6379[1]> zrange test2 0 100  withscores
//    // 1) "key3"
//    // 2) "50"
//    // 3) "key1"
//    // 4) "100"
//    // 5) "key2"
//    // 6) "101"
//    spark.sparkContext.toRedisZSET(rdd, "test2")
//
//  }
//
//  def loadFromRedis2(spark: SparkSession): Unit = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    val rdd = sc.fromRedisZSetWithScore("test", 3)
//    val df = rdd.toDF("mac", "timestamp")
//    df.printSchema()
//    df.show(false)
//
//    val df2 = df.withColumn("timestamp", col("timestamp").cast(LongType))
//    df2.show()
//
//    println(df2.rdd.getNumPartitions)
//    val df3 = df2.rdd.mapPartitionsWithIndex((partId, iter) => {
//      // 这里foreach打印之后，就不能在后面再show了，会无数据
//      //      iter.foreach(x => {
//      //        println("partitionId: " + partId + " value: " + Tuple2())
//      //      })
//      iter
//        .map(x => basic.Value(x.getAs[String]("mac"), x.getAs[Long]("timestamp"), partId))
//        .toList
//        .sortBy(x => x.timestamp)
//        .iterator
//    })
//      .toDF()
//    df3.show()
//
//    println("partition 0")
//    println(df3.filter(col("partitionId") === lit(0)).count())
//    println("partition 1")
//    println(df3.filter(col("partitionId") === lit(1)).count())
//    println("partition 2")
//    println(df3.filter(col("partitionId") === lit(2)).count())
//  }
//
//
//  def loadFromRedis(spark: SparkSession): Unit = {
//    import com.redislabs.provider.redis._
//    import spark.implicits._
//    val sc = spark.sparkContext
//
//    // 这种方式只会得到值，而拿不到分数
//    //    val rdd = sc.fromRedisZRange("south:stb:online.info", 0, 1696089600, 3)
//    val rdd = sc.fromRedisZSetWithScore("south:stb:online.info", 3)
//    val df = rdd.toDF("mac", "timestamp")
//    df.printSchema()
//    df.show(false)
//
//    val df2 = df.withColumn("timestamp", col("timestamp").cast(LongType)).repartition(2)
//    df2.show()
//
//
//    println(df2.rdd.getNumPartitions)
//    df2.rdd.mapPartitionsWithIndex((partId, iter) => {
//      // 这里foreach打印之后，就不能在后面再show了，会无数据
////      iter.foreach(x => {
////        println("partitionId: " + partId + " value: " + Tuple2())
////      })
//      iter
//        .map(x => basic.Value(x.getAs[String]("mac"), x.getAs[Long]("timestamp"), partId))
//        .toList
//        .sortBy(x => x.timestamp)
//        .iterator
//    })
//      .toDF()
//      .show()
//  }
//
//}
//
//case class Value(var mac: String,
//                 var timestamp: Long,
//                 var partitionId: Int)
//
//object Value {
//  def apply(t: Row, partId: Int): basic.Value = {
//    basic.Value(
//      t.getAs[String]("mac"),
//      t.getAs[Long]("timestamp"),
//      partId
//    )
//  }
//}